package cn.demoncat.util.kafka.config;

import cn.demoncat.util.exception.ReflexRuntimeException;
import cn.demoncat.util.mq.util.MqUtil;
import cn.demoncat.util.web.annotation.WebApp;
import org.reflections.Reflections;
import org.reflections.scanners.Scanners;
import org.reflections.util.ConfigurationBuilder;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;
import org.springframework.stereotype.Component;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Map;

/**
 * Bean初始化后置处理器 - 修改@KafkaListener的topics
 *
 * @see KafkaListenerAnnotationBeanPostProcessor
 *
 * @author 延晓磊
 * @since 2023年04月24日
 */
@Component
public class KafkaListenerBeanPostProcessor implements BeanFactoryPostProcessor {

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        // 扫描包
        for (String packageName : WebApp.class.getAnnotation(SpringBootApplication.class).scanBasePackages()) {
            // 读取注解方法
            for (Method method : new Reflections(new ConfigurationBuilder().forPackages(packageName).addScanners(Scanners.MethodsAnnotated)).getMethodsAnnotatedWith(KafkaListener.class)) {
                try {
                    // 获取注解，变更主题
                    changeTopics(method.getAnnotation(KafkaListener.class));
                } catch (Exception e) {
                    throw new ReflexRuntimeException("添加Kafka主题前缀失败", e);
                }
            }
        }
    }

    /**
     * 变更主题
     *
     * @param annotation
     * @throws Exception
     */
    private void changeTopics(KafkaListener annotation) throws Exception {
        // 获取代理
        InvocationHandler invocationHandler = Proxy.getInvocationHandler(annotation);
        // 获取字段
        Field memberValuesField = invocationHandler.getClass().getDeclaredField("memberValues");
        memberValuesField.setAccessible(true);
        Map<String,Object> memberValues = (Map<String, Object>) memberValuesField.get(invocationHandler);
        // 修改字段
        String[] topics = (String[]) memberValues.get("topics");
        for (int i = 0; i < topics.length; i++) {
            // 添加主题前缀
            topics[i] = MqUtil.getTopicExpression(topics[i]);
        }
    }
}
